package drds.binlog.server.netty;

import drds.binlog.protocol.protocol_proto.Ack;
import drds.binlog.protocol.protocol_proto.Packet;
import drds.binlog.protocol.protocol_proto.PacketType;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.buffer.CompositeChannelBuffer;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.util.HashedWheelTimer;
import org.jboss.netty.util.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.ArrayList;
import java.util.List;

public class Nettys
{

    private static final Logger logger = LoggerFactory.getLogger(Nettys.class);
    public static int HEADER_LENGTH = 4;
    public static Timer hashedWheelTimer = new HashedWheelTimer();

    public static void write(Channel channel, ByteBuffer byteBuffer, ChannelFutureListener channelFutureListner)
    {
        byte[] header = ByteBuffer.allocate(HEADER_LENGTH).order(ByteOrder.BIG_ENDIAN).putInt(byteBuffer.limit()).array();
        List<ChannelBuffer> components = new ArrayList<ChannelBuffer>(2);
        components.add(ChannelBuffers.wrappedBuffer(ByteOrder.BIG_ENDIAN, header));
        components.add(ChannelBuffers.wrappedBuffer(byteBuffer));

        if (channelFutureListner == null)
        {
            Channels.write(channel, new CompositeChannelBuffer(ByteOrder.BIG_ENDIAN, components));
        } else
        {
            Channels.write(channel, new CompositeChannelBuffer(ByteOrder.BIG_ENDIAN, components))
                    .addListener(channelFutureListner);
        }
    }

    public static void write(Channel channel, byte[] body, ChannelFutureListener channelFutureListner)
    {
        byte[] header = ByteBuffer.allocate(HEADER_LENGTH).order(ByteOrder.BIG_ENDIAN).putInt(body.length).array();
        if (channelFutureListner == null)
        {
            Channels.write(channel, ChannelBuffers.wrappedBuffer(header, body));
        } else
        {
            Channels.write(channel, ChannelBuffers.wrappedBuffer(header, body)).addListener(channelFutureListner);
        }
    }

    public static void ack(Channel channel, ChannelFutureListener channelFutureListner)
    {
        write(channel,
                Packet.newBuilder()
                        .setType(PacketType.ack)
                        .setBody(Ack.newBuilder().build().toByteString())
                        .build()
                        .toByteArray(),
                channelFutureListner);
    }

    public static void error(int errorCode, String errorMessage, Channel channel,
                             ChannelFutureListener channelFutureListener)
    {
        if (channelFutureListener == null)
        {
            channelFutureListener = ChannelFutureListener.CLOSE;
        }

        logger.error("ErrotCode:{} , Caused by : \n{}", errorCode, errorMessage);
        write(channel,
                Packet.newBuilder()
                        .setType(PacketType.ack)
                        .setBody(Ack.newBuilder().setErrorCode(errorCode).setErrorMessage(errorMessage).build().toByteString())
                        .build()
                        .toByteArray(),
                channelFutureListener);
    }

    public static byte[] ackPacket()
    {
        return Packet.newBuilder()
                .setType(PacketType.ack)
                .setBody(Ack.newBuilder().build().toByteString())
                .build()
                .toByteArray();
    }

    public static byte[] errorPacket(int errorCode, String errorMessage)
    {
        return Packet.newBuilder()
                .setType(PacketType.ack)
                .setBody(Ack.newBuilder().setErrorCode(errorCode).setErrorMessage(errorMessage).build().toByteString())
                .build()
                .toByteArray();
    }
}
